package com.mjf.day8

import java.sql.Timestamp

import com.mjf.day3.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.{Over, Table, Tumble}
import org.apache.flink.table.api.scala._
import org.apache.flink.types.Row

/**
 * TableApi时间语义
 *   .proctime处理时间
 *   .rowtime事件时间
 */
object WindowExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    // 指定时间语义为EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 读取数据到DataStream
    val inputStream: DataStream[SensorReading] = env
      .readTextFile("D:\\coding\\idea\\flink-stu\\src\\main\\input\\sensor.txt")
      .map{
        data =>
        val dataArray: Array[String] = data.split(",")
          SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      }
      // 定义watermark
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L
      })

    // 创建表执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // 基于DataStream数据流，转换为一张表，然后进行操作
    val sensorTable: Table =
      tableEnv.fromDataStream(inputStream, 'id, 'timestamp.rowtime as 'ts, 'temperature)

    // 窗口操作
    // 1.1 Group 窗口，开一个10秒滚动窗口，统计每一个传感器温度的数量
    val resultTable: Table = sensorTable
      .window(Tumble over 10.seconds on 'ts as 'tw)
      .groupBy('tw, 'id)
      .select('id, 'id.count, 'tw.end)

    resultTable.printSchema()
    resultTable.toAppendStream[(String, Long, Timestamp)].print("group result")

    // 1.2 Group 窗口，SQL实现
    val resultSqlTable: Table = tableEnv.sqlQuery(
      s"""
         |select
         |   id,
         |   count(id),
         |   tumble_end(ts, interval '10' second)
         |from ${sensorTable}
         |group by
         |   id,
         |   tumble(ts, interval '10' second)
         |""".stripMargin)

    resultSqlTable.printSchema()
    resultSqlTable.toAppendStream[Row].print("group sql result")


    // 2.1 Over 窗口，对每个传感器统计每一行数据与前两行数据的平均温度
    val overResultTable: Table = sensorTable
      .window(Over partitionBy 'id orderBy 'ts preceding 2.rows as 'ow)
      .select('id, 'ts, 'id.count over 'ow, 'temperature.avg over 'ow)

    overResultTable.toAppendStream[Row].print("over result")

    // 2.2 Over 窗口，SQL实现
    val overSqlResultTable: Table = tableEnv.sqlQuery(
      s"""
        |select
        |   id,
        |   ts,
        |   count(id) over w,
        |   avg(temperature) over w
        |from ${sensorTable}
        |window w as (partition by id order by ts rows between 2 preceding and current row)
        |""".stripMargin)

    overSqlResultTable.toAppendStream[Row].print("over sql result")


    env.execute("WindowExample")

  }
}
